package com.cctc.rds.nrpc.core.client.handler;

import com.cctc.rds.nrpc.core.client.connection.ClientConnectionStatus;
import com.cctc.rds.nrpc.core.client.helper.ChannelPromiseHolder;
import com.cctc.rds.nrpc.core.protocol.message.impl.HeartBeatMessage;
import com.cctc.rds.nrpc.core.protocol.message.impl.RpcResponseMessage;
import com.cctc.rds.nrpc.core.util.PidUtils;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.AttributeKey;

import java.util.concurrent.atomic.AtomicInteger;

public class ClientRpcMessageHandler extends ChannelInboundHandlerAdapter {
    private AtomicInteger status;
    private Runnable failRetryConnectCallbale;

    public ClientRpcMessageHandler(AtomicInteger status, Runnable failRetryConnectCallbale) {
        this.status = status;
        this.failRetryConnectCallbale = failRetryConnectCallbale;
    }

    public int getStatus() {
        return status.get();
    }

    public boolean hasStarted() {
        return getStatus() == ClientConnectionStatus.STARTED.getValue();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        status.set(ClientConnectionStatus.STARTED.getValue());
        sendHeartBeat(ctx);
    }

    private void sendHeartBeat(ChannelHandlerContext ctx) {
        HeartBeatMessage heartbeat = HeartBeatMessage.builder().pid(PidUtils.getPid()).build();
        ctx.writeAndFlush(heartbeat);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof HeartBeatMessage) {
            HeartBeatMessage heartbeat = (HeartBeatMessage) msg;
            if (ctx.channel().attr(AttributeKey.valueOf("PID")).get() == null) {
                ctx.channel().attr(AttributeKey.valueOf("PID")).set(heartbeat.getPid());
            }
        } else if (msg instanceof RpcResponseMessage) {
            RpcResponseMessage response = (RpcResponseMessage) msg;
            ChannelPromiseHolder.completePromise(response.getTxid(), response);
        }
    }

    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        status.set(ClientConnectionStatus.OFF.getValue());
        failRetryConnectCallbale.run();
    }
}
